package com.atguigu.edu.realtime.dwd.db.split.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.TableProcessDwd;
import com.atguigu.edu.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;


public class BaseDbTableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>> {

    private MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor;
    private Map<String, TableProcessDwd> configMap = new HashMap<>();

    //声明构造函数
    public BaseDbTableProcessFunction(MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    // TODO: 2024/10/19 首先得把配置表的配置信息    先预加载到程序当中   避免主流业务数据先到  广播流后到   丢数据的情况
    @Override
    public void open(Configuration parameters) throws Exception {
        Connection mysqlConnections = JdbcUtil.getMysqlConnection();
        String sql = "select * from edu0422_config.table_process_dwd";
        List<TableProcessDwd> tableProcessDwdList = JdbcUtil.queryList(mysqlConnections, sql, TableProcessDwd.class, true);
        for (TableProcessDwd tableProcessDwd : tableProcessDwdList) {
            String sourceTable = tableProcessDwd.getSourceTable();
            String sourceType = tableProcessDwd.getSourceType();
            String key = getKey(sourceTable, sourceType);
            configMap.put(key,tableProcessDwd);
        }
        JdbcUtil.closeMysqlConnection(mysqlConnections);
    }

    // TODO: 2024/10/19 拼接key
    private String getKey(String sourceTable, String sourceType) {
        return sourceTable +":"+sourceType;
    }

    // TODO: 2024/10/19 处理主流业务数据
    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {
        //1.获取广播状态
        ReadOnlyBroadcastState<String, TableProcessDwd> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //2.获取操作的业务数据库表的表名
        String table = jsonObject.getString("table");
        //3.获取对业务数据库表操作的类型
        String type = jsonObject.getString("type");
        //4.拼接key
        String key = getKey(table, type);
        //5.根据key   到广播状态以及configMap获取对应的配置信息
        TableProcessDwd tableProcessDwd =null;

        if ((tableProcessDwd = broadcastState.get(key))!= null
                || (tableProcessDwd =configMap.get(key))!= null){
            // 1.如果进入该分支    说明是需要进行动态分流的事实表
            JSONObject dataJsonObj = jsonObject.getJSONObject("data");
            // 在向下游传递之前     过滤掉不需要传递的属性
            String sinkColumns = tableProcessDwd.getSinkColumns();
            deleteNotNeedColumn(dataJsonObj,sinkColumns);
            dataJsonObj.put("ts",dataJsonObj.getLong("ts"));
            collector.collect(Tuple2.of(dataJsonObj,tableProcessDwd));
        }


    }

    private void deleteNotNeedColumn(JSONObject dataJsonObj, String sinkColumns) {
        List<String> columnList = Arrays.asList(sinkColumns.split(","));
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry -> !columnList.contains(entry.getKey()));
    }

    // TODO: 2024/10/9 处理广播流 配置信息
    @Override
    public void processBroadcastElement(TableProcessDwd tableProcessDwd, BroadcastProcessFunction<JSONObject, TableProcessDwd, Tuple2<JSONObject, TableProcessDwd>>.Context context, Collector<Tuple2<JSONObject, TableProcessDwd>> collector) throws Exception {
        // 获取对配置表的操作类型
        String op = tableProcessDwd.getOp();
        // 获取广播状态
        BroadcastState<String, TableProcessDwd> broadcastState =
                context.getBroadcastState(mapStateDescriptor);
        String sourceTable = tableProcessDwd.getSourceTable();
        String sourceType = tableProcessDwd.getSourceType();
        String key = getKey(sourceTable, sourceType);
        if ("d".equals(op)){
            //说明我们从配置表删除了一条配置信息     需要将配置信息从广播状态以及configMap中删除掉
            broadcastState.remove(key);
            configMap.remove(key);
        }else {
            //说明我们从配置表中读/增/改了一条配置   需要将最新配置信息放到广播状态以及configMap中
            broadcastState.put(key,tableProcessDwd);
            configMap.put(key,tableProcessDwd);
        }
    }
}
